  WITH -- 所有车次
       branch_time_effective AS (-- 前一天及当天车次
                                 SELECT depart_network_code,
                                        arrive_network_code,
                                        timestamp(dt || ' ' || latest_collect_time || ':00+08:00') AS latest_collect_time,
                                        timestamp(string(
                                            date_add(dt,
                                                     int(span_days_arrive) + -- 跨天运输
                                                     if(latest_collect_time > planned_departure_time, 1, 0) -- 次日班次
                                                )) || ' ' || planned_arrival_time || ':00+08:00') AS planned_arrival_time
                                   FROM jms_ods.yl_tms_branch_time_effective
                                  WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -1) AND '{{ execution_date | cst_ds }}'
                                    AND is_enable = 1 -- 启用状态
                                    AND line_type = 1 -- 集货线路
                                    AND dt BETWEEN effective_date AND expiration_date -- 有效期内
                                  UNION ALL
                                 -- 次日车次：用于匹配揽件时间过晚，当日已无车次的情况（如果次日分区为空，则在当日车次的基础上 +1 天，及默认次日的班次与当日一致）
                                 SELECT depart_network_code,
                                        arrive_network_code,
                                        timestamp(string(
                                            date_add(dt, if(dt == '{{ execution_date | cst_ds }}', 1, 0) -- 由当天排班虚拟生成次日排班
                                                )) || ' ' || latest_collect_time || ':00+08:00') AS latest_collect_time,
                                        timestamp(string(
                                            date_add(dt,
                                                     int(span_days_arrive) + -- 跨天运输
                                                     if(latest_collect_time > planned_departure_time, 1, 0) + -- 次日班次
                                                     if(dt == '{{ execution_date | cst_ds }}', 1, 0) -- 由当天排班虚拟生成次日排班
                                                )) || ' ' || planned_arrival_time || ':00+08:00') AS planned_arrival_time
                                   FROM jms_ods.yl_tms_branch_time_effective
                                  WHERE dt = date_add('{{ execution_date | cst_ds }}',
                                      if((SELECT TRUE FROM jms_ods.yl_tms_branch_time_effective WHERE dt = date_add('{{ execution_date | cst_ds }}', 1) LIMIT 1), 1, 0))
                                    AND is_enable = 1 -- 启用状态
                                    AND line_type = 1 -- 集货线路
                                    AND dt BETWEEN effective_date AND expiration_date), -- 有效期内
       -- 对车次排序，得到每趟车次对应的揽件时间范围，便于后续通过揽件时间匹配到最近的下一班车次
       planned_line AS (SELECT *,
                               nvl(lead(latest_collect_time) OVER w, timestamp(date(latest_collect_time))) AS prev_collect_time
                          FROM branch_time_effective
                        WINDOW w AS (PARTITION BY depart_network_code, arrive_network_code ORDER BY latest_collect_time DESC, planned_arrival_time)),
       -- 集货 distribution
       collect_distribution AS (SELECT network_id        AS network_id, -- 源网点编码
                                       distribution_type AS distribution_type, -- 2: 网点/集散交中心 4: 网点交集散
                                       collect_set(code) AS distribution_codes -- 可集货目标网点编码
                                  FROM jms_ods.yl_lmdm_sys_network_distributi
                                 WHERE dt = '{{ execution_date | cst_ds }}'
                                   AND distribution_type IN (2, 4)
                                 GROUP BY distribution_type, network_id),
       -- 订单细分来源，取创建时间最早的渠道记录
       sub_source AS (SELECT waybill_no, sub_source_code
                        FROM (SELECT way_bill_no  AS waybill_no,
                                     order_source AS sub_source_code,
                                     row_number() OVER (PARTITION BY way_bill_no ORDER BY create_time) AS rk
                                FROM jms_ods.tab_statistics
                               WHERE dt BETWEEN date_add('{{ execution_date | cst_ds }}', -4) AND '{{ execution_date | cst_ds }}')
                       WHERE rk = 1),
       -- 集货流程
       transfer AS (SELECT trace.waybill_no                                                                                                                                                 AS waybill_no,
                           first_value(if(trace.step = 1, trace.est_scan_time, NULL), TRUE)                                                                                                 AS collect_start_time,
                           first_value(if(trace.step = 1, trace.network_code, NULL), TRUE)                                                                                                  AS taking_network_code,
                           first_value(if(trace.next_network_is_est_entrepot, trace.next_network_code, NULL), TRUE)                                                                         AS entrepot_network_code,
                           first_value(if(trace.next_network_is_est_entrepot, trace.load_shipment_departure_time, NULL), TRUE)                                                              AS entrepot_departure_time,
                           first_value(if(trace.next_network_is_est_entrepot, nvl(trace.next_unload_shipment_arrival_time, trace.next_est_scan_time), NULL), TRUE)                          AS entrepot_arrival_time,
                           first_value(if(trace.next_network_is_est_center, trace.load_shipment_departure_time, NULL), TRUE)                                                                AS center_departure_time,
                           first_value(if(trace.next_network_is_est_center, nvl(trace.next_unload_shipment_arrival_time, trace.next_est_scan_time), NULL), TRUE)                            AS center_arrival_time,
                           first_value(if(trace.next_network_is_est_center, trace.next_network_code, NULL), TRUE)                                                                           AS center_code,
                           first_value(if(trace.next_network_is_est_entrepot, nvl(plan.planned_arrival_time, timestamp(date_add(trace.est_scan_time, 1)) - INTERVAL 1 SECOND), NULL), TRUE) AS entrepot_planned_arrival_time
                      FROM jms_dm.dm_waybill_trace_dt AS trace
                      LEFT JOIN planned_line      AS plan
                        ON trace.next_network_code IS NOT NULL
                       AND plan.arrive_network_code = trace.next_network_code
                       AND plan.depart_network_code = trace.network_code
                       AND trace.est_scan_time > plan.prev_collect_time
                       AND trace.est_scan_time <= plan.latest_collect_time
                     WHERE trace.dt BETWEEN date_add('{{ execution_date | cst_ds }}', -1) AND '{{ execution_date | cst_ds }}'
                       AND trace.waybill_no LIKE 'JT_____________'
                       -- 从普通网点始发，直接到达中心或经过集散点到达中心
                       AND ((trace.step = 1 AND (trace.network_is_center OR trace.network_is_entrepot) = FALSE AND trace.next_network_is_est_entrepot OR trace.next_network_is_est_center) OR
                            (trace.step = 2 AND trace.network_is_entrepot = TRUE AND trace.next_network_is_est_center))
                       AND (trace.est_center_step <= 3 OR trace.est_center_step IS NULL)
                     GROUP BY waybill_no
                    HAVING first_value(if(trace.step = 1, date(trace.est_scan_time), NULL), TRUE) = date_add('{{ execution_date | cst_ds }}', -1)),
       -- 揽收点/集散 -> 中心 计划
       transfer_with_plan AS (SELECT *, nvl(plan.planned_arrival_time, timestamp(date_add(transfer.collect_start_time, 1)) - INTERVAL 1 SECOND) AS center_planned_arrival_time
                                FROM transfer
                                LEFT JOIN planned_line      AS plan
                                  ON transfer.center_code IS NOT NULL
                                 AND plan.arrive_network_code = transfer.center_code
                                 AND plan.depart_network_code = nvl(transfer.entrepot_network_code, transfer.taking_network_code)
                                 AND if(transfer.entrepot_network_code IS NOT NULL, transfer.entrepot_planned_arrival_time, transfer.collect_start_time) > plan.prev_collect_time
                                 AND if(transfer.entrepot_network_code IS NOT NULL, transfer.entrepot_planned_arrival_time, transfer.collect_start_time) <= plan.latest_collect_time),
       -- 关联信息、计算是否准点
       calc_is_in_time AS (SELECT transfer.*,
                                  CASE WHEN entrepot_arrival_time IS NOT NULL                                            THEN entrepot_arrival_time <= entrepot_planned_arrival_time
                                       WHEN date_add('{{ execution_date | cst_ds }}', 1) > entrepot_planned_arrival_time THEN FALSE END AS is_entrepot_in_time, -- 到达集散是否准点
                                  CASE WHEN center_arrival_time IS NOT NULL                                            THEN center_arrival_time <= center_planned_arrival_time
                                       WHEN date_add('{{ execution_date | cst_ds }}', 1) > center_planned_arrival_time THEN FALSE END AS is_center_in_time,    -- 到达中心是否准点
                                  waybill.goods_type_id      AS goods_type_id,
                                  waybill.goods_type_code    AS goods_type_code,
                                  waybill.goods_type_name    AS goods_type_name,
                                  odr.order_source_code      AS order_source_code,
                                  odr.order_source_name      AS order_source_name,
                                  sub_source.sub_source_code AS sub_source_code,
                                  taking.id                  AS taking_network_id,
                                  entrepot.id                AS entrepot_network_id,
                                  entrepot.name              AS entrepot_network_name,
                                  center.id                  AS center_id
                             FROM transfer_with_plan AS transfer
                             -- 揽收网点
                             LEFT JOIN jms_ods.yl_lmdm_sys_network AS taking
                               ON taking.dt = '{{ execution_date | cst_ds }}'
                              AND taking.code = transfer.taking_network_code
                             -- 集散点
                             LEFT JOIN jms_ods.yl_lmdm_sys_network AS entrepot
                               ON entrepot.dt = '{{ execution_date | cst_ds }}'
                              AND entrepot.code = transfer.entrepot_network_code
                             -- 中心
                             LEFT JOIN jms_ods.yl_lmdm_sys_network AS center
                               ON center.dt = '{{ execution_date | cst_ds }}'
                              AND center.code = transfer.center_code
                             -- 运单信息
                             LEFT JOIN jms_dwd.dwd_yl_oms_oms_waybill_dt AS waybill
                               ON waybill.dt BETWEEN date_add('{{ execution_date | cst_ds }}', -2) AND '{{ execution_date | cst_ds }}'
                              AND waybill.waybill_no = transfer.waybill_no
                             -- 订单信息: 拼多多 D09/D60
                             LEFT JOIN jms_dwd.dwd_yl_oms_oms_order_dt AS odr
                               ON odr.dt BETWEEN date_add('{{ execution_date | cst_ds }}', -4) AND '{{ execution_date | cst_ds }}'
                              AND odr.waybill_id = transfer.waybill_no
                             -- 订单细分来源: 1085 - 抖音, 1133 - 快手
                             LEFT JOIN sub_source ON transfer.waybill_no = sub_source.waybill_no
                             -- 完结件不做统计
                             LEFT ANTI JOIN jms_ods.tab_end_piece AS end_piece
                               ON end_piece.dt = '{{ execution_date | cst_ds }}'
                              AND end_piece.end_code IN (30, 44, 41, 5) -- 30: 违禁品, 44: 作废件, 41: 遗失件, 5: 客户取消寄件
                              AND end_piece.input_time BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND date_add('{{ execution_date | cst_ds }}', 1)
                              AND end_piece.waybill_no = transfer.waybill_no
                              -- 完结时间早于计划到达时间
                              AND end_piece.input_time <= transfer.center_planned_arrival_time
                             -- 拦截件不做统计（取消订单、错分件、问题件、退回件、转寄件）
                             LEFT ANTI JOIN jms_ods.yl_oms_interceptorpiece AS interceptor_piece
                               ON interceptor_piece.dt = '{{ execution_date | cst_ds }}'
                              AND interceptor_piece.registration_time BETWEEN date_add('{{ execution_date | cst_ds }}', -6) AND date_add('{{ execution_date | cst_ds }}', 1)
                              AND interceptor_piece.waybill_no = transfer.waybill_no
                              -- 拦截时间早于计划到达时间
                              AND interceptor_piece.registration_time <= transfer.center_planned_arrival_time
                            WHERE
                             -- 剔除 退件、转寄件、废件、问题件
                             nvl(waybill.is_refund, 0) + nvl(waybill.is_void, 0) + nvl(waybill.is_abnormal, 0) = 0),
       -- 关联交件关系
       calc_distribution_relation AS (SELECT calc_is_in_time.*,
                                             nvl(array_contains(entrepot_distribution.distribution_codes, entrepot_network_code), TRUE) AS is_to_entrepot_conform_with_distribution, -- 揽收点到集散点是否可交件
                                             nvl(array_contains(center_distribution.distribution_codes, center_code), TRUE)             AS is_to_center_conform_with_distribution    -- 揽件点/集散点到中心是否可交件
                                        FROM calc_is_in_time
                                        LEFT JOIN collect_distribution AS entrepot_distribution ON entrepot_network_id IS NOT NULL AND entrepot_distribution.distribution_type = 4 AND taking_network_id = entrepot_distribution.network_id
                                        LEFT JOIN collect_distribution AS center_distribution ON center_id IS NOT NULL AND center_distribution.distribution_type = 2 AND nvl(entrepot_network_id, taking_network_id) = center_distribution.network_id),
       -- 计算责任方
       calc_duty AS (SELECT *,
                            CASE WHEN least(is_to_entrepot_conform_with_distribution, is_to_center_conform_with_distribution) = FALSE THEN NULL                  -- 实际交件集散/中心 与 可交件集散/中心 不一致
                                 WHEN is_center_in_time AND NOT is_entrepot_in_time                                                   THEN NULL                  -- 中心准点，集散迟到
                                 WHEN is_center_in_time AND nvl(is_entrepot_in_time, TRUE)                                            THEN NULL                  -- 中心准点，集散未到或准点
                                 WHEN NOT nvl(is_center_in_time, FALSE) AND NOT nvl(is_entrepot_in_time, FALSE)                       THEN taking_network_code   -- 中心未到或迟到，集散未到或迟到
                                 WHEN NOT nvl(is_center_in_time, FALSE) AND is_entrepot_in_time                                       THEN entrepot_network_code END -- 中心未到或迟到，集散准点
                                                                                                                                      AS duty_network_code
                       FROM calc_distribution_relation)
INSERT OVERWRITE TABLE jms_dm.dm_waybill_collect_aging_dt PARTITION (dt='{{ execution_date | date_add(-1) | cst_ds }}')
SELECT transfer.waybill_no                                                AS waybill_no,                    -- 运单号
       transfer.collect_start_time                                        AS collect_start_time,            -- 揽收点最早扫描时间，揽收/入仓/发件的最早时间
       taking_network.regional_id                                         AS taking_regional_id,            -- 收件大区ID
       taking_network.regional_desc                                       AS taking_regional_desc,          -- 收件大区名称
       taking_network.provider_id                                         AS taking_provider_id,            -- 收件省份ID
       taking_network.provider_desc                                       AS taking_provider_desc,          -- 收件省份名称
       taking_network.city_id                                             AS taking_city_id,                -- 收件城市ID
       taking_network.city_desc                                           AS taking_city_desc,              -- 收件城市名称
       taking_network.subordinate_agent_id                                AS taking_agent_id,               -- 收件代理区ID
       taking_network.subordinate_agent_code                              AS taking_agent_code,             -- 收件代理区编码
       taking_network.subordinate_agent_name                              AS taking_agent_name,             -- 收件代理区名称
       taking_network.first_franchisee_id                                 AS taking_first_franchisee_id,    -- 收件一级加盟商ID
       taking_network.first_franchisee_code                               AS taking_first_franchisee_code,  -- 收件一级加盟商编码
       taking_network.first_franchisee_name                               AS taking_first_franchisee_name,  -- 收件一级加盟商名称
       taking_network.second_franchisee_id                                AS taking_second_franchisee_id,   -- 收件二级加盟商ID
       taking_network.second_franchisee_code                              AS taking_second_franchisee_code, -- 收件二级加盟商编码
       taking_network.second_franchisee_name                              AS taking_second_franchisee_name, -- 收件二级加盟商名称
       taking_network.id                                                  AS taking_network_id,             -- 收件网点ID
       taking_network.code                                                AS taking_network_code,           -- 收件网点编码
       taking_network.name                                                AS taking_network_name,           -- 收件网点名称
       transfer.entrepot_network_id                                       AS entrepot_network_id,           -- 交件集散点ID
       transfer.entrepot_network_code                                     AS entrepot_network_code,         -- 交件集散点编码
       transfer.entrepot_network_name                                     AS entrepot_network_name,         -- 交件集散点名称
       transfer.entrepot_departure_time                                   AS entrepot_departure_time,       -- 至集散点的实际发车时间
       transfer.entrepot_arrival_time                                     AS entrepot_arrival_time,         -- 至集散点的实际到车时间
       transfer.entrepot_planned_arrival_time                             AS entrepot_planned_arrival_time, -- 至集散点的规划到达时间
       transfer.is_entrepot_in_time                                       AS is_entrepot_in_time,           -- 到达集散是否准点
       center.id                                                          AS center_id,                     -- 交件中心ID
       center.code                                                        AS center_code,                   -- 交件中心编码
       center.name                                                        AS center_name,                   -- 交件中心名称
       center.subordinate_agent_id                                        AS center_agent_id,               -- 交件中心代理区ID
       center.subordinate_agent_code                                      AS center_agent_code,             -- 交件中心代理区编码
       center.subordinate_agent_name                                      AS center_agent_name,             -- 交件中心代理区名称
       transfer.center_departure_time                                     AS center_departure_time,         -- 至交件中心的实际发车时间
       transfer.center_arrival_time                                       AS center_arrival_time,           -- 至交件中心的实际到车时间
       transfer.center_planned_arrival_time                               AS center_planned_arrival_time,   -- 至交件中心的规划到达时间
       transfer.is_center_in_time                                         AS is_center_in_time,             -- 到达中心是否准点
       taking_network.subordinate_agent_id != center.subordinate_agent_id AS is_diff_agent,                 -- 是否跨代理区
       CASE WHEN transfer.duty_network_code = transfer.taking_network_code   THEN transfer.taking_network_id
            WHEN transfer.duty_network_code = transfer.entrepot_network_code THEN transfer.entrepot_network_id
           END                                                            AS duty_network_id,               -- 交件不准点责任所属网点ID
       transfer.duty_network_code                                         AS duty_network_code,             -- 交件不准点责任所属网点编码
       CASE WHEN transfer.duty_network_code = transfer.taking_network_code   THEN taking_network.name
            WHEN transfer.duty_network_code = transfer.entrepot_network_code THEN transfer.entrepot_network_name
           END                                                            AS duty_network_name,             -- 交件不准点责任所属网点名称
       transfer.goods_type_id                                             AS goods_type_id,                 -- 物品类型ID
       transfer.goods_type_code                                           AS goods_type_code,               -- 物品类型code
       transfer.goods_type_name                                           AS goods_type_name,               -- 物品类型名称
       transfer.order_source_code                                         AS order_source_code,             -- 渠道code
       transfer.order_source_name                                         AS order_source_name,             -- 渠道名称
       transfer.sub_source_code                                           AS sub_source_code,               -- 渠道细分code
       transfer.is_to_entrepot_conform_with_distribution                  AS is_to_entrepot_conform_with_distribution, -- 交件到集散点是否符合交件关系
       transfer.is_to_center_conform_with_distribution                    AS is_to_center_conform_with_distribution    -- 交件到中心是否符合交件关系
  FROM calc_duty                               AS transfer
  LEFT JOIN jms_ods.yl_lmdm_sys_network_expand AS taking_network ON taking_network.dt = '{{ execution_date | cst_ds }}' AND transfer.taking_network_code = taking_network.code
  LEFT JOIN jms_ods.yl_lmdm_sys_network_expand AS center ON center.dt = '{{ execution_date | cst_ds }}' AND transfer.center_code IS NOT NULL AND transfer.center_code = center.code
DISTRIBUTE BY pmod(hash(waybill_no), 10);
